package com.flute.haflute.agent.job;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.Map;

import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IoSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.agent.topo.Node;
import com.flute.haflute.common.ClusterConstants;
import com.flute.haflute.common.ClusterContext;
import com.flute.haflute.jobbox.base.CloudJobConf;
import com.flute.haflute.net.messaging.Message;
import com.flute.haflute.net.messaging.MessageFactory;
import com.flute.haflute.net.messaging.MessagingService;
import com.flute.haflute.net.messaging.Message.MessageType;
import com.flute.haflute.tools.ClusterUtils;
import com.flute.haflute.tools.Utils;

/**
 * 一项Job的资源上下文建立者<br>
 * 1. 获取实际的Job上下文策略<br>
 * 2. 根据策略获取Job上下文资源的分配情况<br>
 * 3. 根据资源的分配情况，进行资源的通知和分配处理，包括：a. 通知主节点；b.通知从节点<br>
 * 4. 登记备录分配过的资源。<br>
 * 
 */
public class JobContextInstaller {
	private static Logger logger = LoggerFactory
			.getLogger(JobContextInstaller.class);
	private static JobContextStrategy strategy;
	private static final int IS_GLOBAL_CHECK_NODES = 0;
	private static int appointval = -1;

	// WEB的NODE选择项
	private static enum NodeCheckedType {
		RANDOM {
			public Map<NodeCheckedType, Integer> getVal() {
				Map<NodeCheckedType, Integer> map = new HashMap<NodeCheckedType, Integer>();
				map.put(this, -1);
				return map;
			}
		},
		APPOINT {
			public Map<NodeCheckedType, Integer> getVal() {
				Map<NodeCheckedType, Integer> map = new HashMap<NodeCheckedType, Integer>();
				map.put(this, JobContextInstaller.getAppointval());
				return map;
			}
		},
		ALLNODE {
			public Map<NodeCheckedType, Integer> getVal() {
				Map<NodeCheckedType, Integer> map = new HashMap<NodeCheckedType, Integer>();
				map.put(this, -1);
				return map;
			}
		};
		abstract Map<NodeCheckedType, Integer> getVal();

	}

	public static NodeCheckedType nodechecked = NodeCheckedType.RANDOM;

	public static void setNodeChecked(String nodechecked) {
		for (NodeCheckedType nodes : NodeCheckedType.values()) {
			if (nodes.toString().equals(nodechecked)) {
				JobContextInstaller.nodechecked = nodes;
			}
		}

	}

	public static NodeCheckedType getNodeChecked() {
		return nodechecked;
	}

	public static Integer getNodeCheckedval() {
		return nodechecked.getVal().get(nodechecked);
	}

	static {
		strategy = new RandomAwareOfHistoryStrategy();
	}

	/**
	 * 该方法必须作为一个事务性质处理，故是同步的，<br>
	 * 增加功能时注意资源的申请和释放
	 * 
	 * @param request
	 *            工作资源请求
	 * @return 分配给该Job的主节点（控制节点）
	 * @throws Exception
	 */
	public synchronized static Node prepare4NewJob(CloudJobConf request)
			throws Exception {
		int nodesNeeded = request.getTasksNeeded();
		if (nodesNeeded == IS_GLOBAL_CHECK_NODES) {
			if (getNodeChecked() == NodeCheckedType.ALLNODE) {
				nodesNeeded = JobResourceManager.getInstance()
						.getAllFreeNodes().size();
				request.setTasksNeeded(nodesNeeded);
			} else if (getNodeChecked() == NodeCheckedType.APPOINT) {
				nodesNeeded = getAppointval();
				request.setTasksNeeded(nodesNeeded);
			}
		}
		// 获得Job可以申请到的资源
		JobContext jobContext = strategy.getJobContext(request);
		logger.info("new Job resouce is {}", jobContext);

		// 记录所有分配给该Job的资源
		HashMap<Node, JobContext> allocatedNodes = new HashMap<Node, JobContext>(
				jobContext.getJobSlaves().size() + 1);
		allocatedNodes.put(jobContext.getJobMaster(), jobContext);
		for (Node slave : jobContext.getJobSlaves()) {
			allocatedNodes.put(slave, jobContext);
		}

		try {
			// 通知工作主节点准备接受新的Job
			notifyNewJobController(jobContext);
			// 通知相关的从节点准备接受新Job的任务Task
			notifyNewJobSlaves(jobContext);
			// 登记分配过的节点
			JobResourceManager.getInstance().registerAllocatedNodes(
					allocatedNodes);
		} catch (Exception e) {
			JobResourceManager.getInstance().releaseAllocatedNodes(
					allocatedNodes);
			throw e;
		}

		return jobContext.getJobMaster();
	}

	/**
	 * 通知相关的从节点准备接受新Job的任务Task
	 * 
	 * @throws IOException
	 */
	private static void notifyNewJobSlaves(JobContext jobContext)
			throws IOException {
		int totalNodes = jobContext.getJobSlaves().size();
		NewTaskMessageResponseHandler responseHandler = new NewTaskMessageResponseHandler(
				totalNodes);
		Message newTaskMessage = MessageFactory.getNewJobMessage();
		newTaskMessage.setHeader(ClusterConstants.JOB_ROLE_TAG,
				Utils.int2bytes(ClusterConstants.JOB_ROLL.CHILD.ordinal()));
		newTaskMessage.setHeader(ClusterConstants.JOB_MASTER_TAG, jobContext
				.getJobMaster().getIp().getBytes());

		logger.info("will send job allocation request to JobSlaves");
		for (Node node : jobContext.getJobSlaves()) {
			MessagingService.getInstance().send2Node(
					new InetSocketAddress(node.getIp(),
							ClusterContext.getHeartBeatPort()), newTaskMessage,
					responseHandler);
		}

		// 等待从节点返回Ok信息
		while (!responseHandler.isReady4All()) {
			ClusterUtils.doWait();
		}
		logger.info("JobSlaves are OK");
	}

	/**
	 * 通知工作主节点准备接受新的Job
	 * 
	 * @param jobContext
	 * @throws IOException
	 */
	private static void notifyNewJobController(JobContext jobContext)
			throws IOException {
		Message newTaskMessage = MessageFactory.getNewJobMessage();
		newTaskMessage.setHeader(ClusterConstants.JOB_ROLE_TAG,
				Utils.int2bytes(ClusterConstants.JOB_ROLL.MASTER.ordinal()));
		// notify job executors
		logger.info("will send job allocation request to JobMaster");
		NewTaskMessageResponseHandler jobMasterRsponseHandler = new NewTaskMessageResponseHandler(
				1);
		MessagingService.getInstance().send2Node(
				new InetSocketAddress(jobContext.getJobMaster().getIp(),
						ClusterContext.getHeartBeatPort()), newTaskMessage,
				jobMasterRsponseHandler);

		// 等待主节点返回Ok信息
		while (!jobMasterRsponseHandler.isReady4All()) {
			ClusterUtils.doWait();
		}
		logger.info("JobMaster is OK");
	}

	/**
	 * 配置工作上下文策略，这在集群工作调度中很重要
	 * 
	 * @param strategy
	 */
	public synchronized static void setJobContextStrategy(
			JobContextStrategy strategy) {
		JobContextInstaller.strategy = strategy;
	}

	/**
	 * 新发任务后监听任务执行从节点返回的消息的处理类 如果返回成功，则设置相关标记
	 * 
	 * @author QiCheng Liu
	 * 
	 */
	private static class NewTaskMessageResponseHandler extends IoHandlerAdapter {
		private int nodes2Notify = 0;
		private int nodesAlreadyNotified = -1;

		public NewTaskMessageResponseHandler(int nodes) {
			this.nodes2Notify = nodes;
			this.nodesAlreadyNotified = 0;
		}

		@Override
		public void messageReceived(IoSession session, Object message)
				throws Exception {
			Message realMessage = (Message) message;
			if (realMessage.getMessageType().equals(MessageType.JOB_NEW.name())) {
				nodesAlreadyNotified++;
				logger.info("OK Response received: {}/{}",
						nodesAlreadyNotified, nodes2Notify);
			}
			if (nodesAlreadyNotified >= nodes2Notify) {
				session.close(true);
			}
		}

		public void exceptionCaught(IoSession session, Throwable cause)
				throws Exception {
			System.err.println("erro session " + session.getId());
			cause.printStackTrace();
		}

		/**
		 * 是否所有任务执行节点都返回成功标记
		 * 
		 * @return
		 */
		public boolean isReady4All() {
			boolean result = (nodesAlreadyNotified >= nodes2Notify);
			return result;
		}
	}

	public static int getAppointval() {
		return appointval;
	}

	public static void setAppointval(int appointval) {
		JobContextInstaller.appointval = appointval;
	}
}
